package com.cjh.wc.fraud;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;

/**
 * FraudDetector 类定义了欺诈检测的业务逻辑
 * @author cjh
 */
public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {
    private static final long serialVersionUID = 1L;

    private static final double SMALL_AMOUNT = 1.00;
    private static final double LARGE_AMOUNT = 500.00;
    private static final long ONE_MINUTE = 60 * 1000;

    /**
     * ValueState是一个包装类，类似于Java标准库中的AtomicReference或AtomicLong。
     * 它提供了三种与内容交互的方法;update设置状态，value获取当前值，clear删除其内容。
     * 如果特定键的状态为空，例如在应用程序开始时或调用ValueState#clear之后，则ValueState#value将返回null。
     * 对ValueState#value返回的对象的修改不能保证被系统识别，因此所有的修改都必须使用ValueState#update执行。
     */
    private transient ValueState<Boolean> flagState;

    private transient ValueState<Long> timerState;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<Boolean> flagDescriptor = new ValueStateDescriptor<>(
                "flag",
                Types.BOOLEAN);
        flagState = getRuntimeContext().getState(flagDescriptor);

        ValueStateDescriptor<Long> timerDescriptor = new ValueStateDescriptor<>(
                "timer-state",
                Types.LONG);
        timerState = getRuntimeContext().getState(timerDescriptor);
    }


    /**
     * 欺诈检测器被部署为KeyedProcessFunction。
     * 这个方法被作用于每个交易事件。
     * 这个第一个版本对每个交易产生一个警告，有些人可能会说这是过于稳健的。
     * 本教程的后续步骤将指导您使用更有意义的业务逻辑扩展欺诈检测器。
     * @param transaction
     * @param context
     * @param collector
     * @throws Exception
     */
    @Override
    public void processElement(
            Transaction transaction,
            Context context,
            Collector<Alert> collector) throws Exception {

        // Get the current state for the current key
        Boolean lastTransactionWasSmall = flagState.value();

        // Check if the flag is set
        if (lastTransactionWasSmall != null) {
            //只有在lastTransactionWasSmall 并且本订单金额大于500时新建Alert
            if (transaction.getAmount() > LARGE_AMOUNT) {
                // Output an alert downstream
                Alert alert = new Alert();
                alert.setId(transaction.getAccountId());

                collector.collect(alert);//只有被collector收集的alert才会流到下一步
            }

            //clean state 当标记状态被重置时，删除定时器。
            cleanUp(context);
        }

        if (transaction.getAmount() < SMALL_AMOUNT) {
            //本金额订单小与1,将flag置为True
            flagState.update(true);

            long timer = context.timerService().currentProcessingTime() + ONE_MINUTE;
            context.timerService().registerProcessingTimeTimer(timer);// 一分钟后调用onTimer,设置一分钟后的触发器

            timerState.update(timer);
        }
    }



    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out) {
        // remove flag after 1 minute
        timerState.clear();
        flagState.clear();
    }

    private void cleanUp(Context ctx) throws Exception {
        // delete timer
        Long timer = timerState.value();
        ctx.timerService().deleteProcessingTimeTimer(timer);//删除定时器,防止定时到的时候误删状态

        // clean up all state
        timerState.clear();
        flagState.clear();
    }
}
